package title4

import java.util.Properties

import org.apache.spark.broadcast.Broadcast
import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}

/**
  * 问题4.构建每一个用户的用户画像，就是根据用户购买的具体商品，给用户打上一个标签，为将来的商品推荐系统作数据支撑
  */
object UserPortrait {
  def main(args: Array[String]): Unit = {
    val session = SparkSession.builder().master("local[*]").appName(this.getClass.getName).getOrCreate()

    import session.implicits._
    //1.读取标签规则数据
    val url = "jdbc:mysql://localhost:3306/scott?characterEncoding=utf-8"
    val tname = "rule"
    val p = new Properties()
    p.setProperty("user", "root")
    p.setProperty("password", "123456")
    p.setProperty("driver", "com.mysql.jdbc.Driver")
    val jdbc = session.read.jdbc(url, tname, p)

    jdbc.createTempView("temptable")
    //2.利用sql列转行，将标签转换成一行数据
    val tempdata = jdbc.sqlContext.sql("select order,concat_ws(',',collect_set(brand)) brands from temptable group by order")

    //3.将列转行的数据作为规则 收集 并广播出去
    val rule: Array[(String, String)] = tempdata.map({ t =>
      val line = t.toString()
      val strings = line.split(",")
      val order: String = strings(0).substring(1)
      val brands: String = line.substring(line.indexOf(",") + 1, line.length - 1)
      (order, brands)
    }).collect()
    rule.foreach(println)
    val ruleBC: Broadcast[Array[(String, String)]] = session.sparkContext.broadcast(rule)


    //4.处理订单数据
    val orderData = session.read.textFile("data/订单数据.log")
    val userAndOrder: DataFrame = orderData.map({ t =>
      val line = t.split(" ")
      val user = line(0)
      val order = line(3)
      (user, order)
    }).toDF("user", "order")
    userAndOrder.createTempView("table")
    //5.查出每个用户的商品 （行转列）
    val userAndOrders: DataFrame = userAndOrder.sqlContext.sql("select user,concat_ws(',',collect_set(order)) order from table group by user")

    //6.根据用户商品匹配标签库
    val result = userAndOrders.map({ t =>
      val rule: Array[(String, String)] = ruleBC.value
      val str = t.toString()
      //用户标识
      val user = str.split(",")(0).substring(1)
      //用户购买的商品
      val orders: Array[String] = str.substring(str.indexOf(",") + 1, str.length - 1).split(",")
      //匹配标签
      val brands: String = searchBrands(orders, rule)
      (user, brands)
    }).toDF("id", "brands")
    result.show()

  }

  /** 结果数据
    * +---+--------------------+
    * | id|              brands|
    * +---+--------------------+
    * |  F|           高端人士, 白富美|
    * |  B|高端人士, 商务男士, 数码一族, 果粉|
    * |  D|                 育儿中|
    * |  C|                 育儿中|
    * |  J|            屌丝, IT人士|
    * |  A|      高端人士, 数码一族, 果粉|
    * |  H|                 育儿中|
    * +---+--------------------+
    * */

  /**
    * 查找商品对应标签
    */
  val searchBrands = (orders: Array[String], rules: Array[(String, String)]) => {

    val ordersIter = orders.iterator
    var result = ""
    while (ordersIter.hasNext) {
      val order = ordersIter.next()
      result += search(order, rules)
    }
    val str = result.substring(0, result.length - 1).split(",").distinct.toBuffer.toString()
    str.substring(12, str.length - 1)
  }
  /**
    * 遍历标签库
    */
  val search = (order: String, rules: Array[(String, String)]) => {
    val ruleIter = rules.iterator
    var result = ""
    while (ruleIter.hasNext) {
      val rule = ruleIter.next()
      if (order.equals(rule._1)) {
        //brands
        result += rule._2 + ","
      } else {
        result += ""
      }
    }
    result
  }
}


